Crud fastapi kết nối với mongodb và upload lên python manager

cài đặt trên python manager chọn famework la python va startup model la gunicorn

# go lenh : /www/wwwroot/fastapi/0d32bced91aa5c2ee5696fc7995370ae_venv/bin/python3 -m pip install -r requirements.txt
# go lenh : /www/wwwroot/fastapi/0d32bced91aa5c2ee5696fc7995370ae_venv/bin/python3 -m pip install  gunicorn
# go lenh : /www/wwwroot/fastapi/0d32bced91aa5c2ee5696fc7995370ae_venv/bin/python3 -m pip install "uvicorn[standard]" gunicorn
#cai dat pip install "uvicorn[standard]" gunicorn
# trong config o giao dien python manager them : worker_class = 'uvicorn.workers.UvicornWorker' thay cho worker_class = "class" nhu sau
# bind = '0.0.0.0:7124'
# user = 'root'
# workers = 1
# threads = 2
# backlog = 512
# chdir = '/www/wwwroot/fastapi'
# access_log_format = '%(t)s %(p)s %(h)s "%(r)s" %(s)s %(L)s %(b)s %(f)s" "%(a)s"'
# loglevel = 'info'
# worker_class = 'uvicorn.workers.UvicornWorker'
# errorlog = chdir + '/logs/error.log'
# accesslog = chdir + '/logs/access.log'
# pidfile = chdir + '/logs/fastapi.pid'

Mình có file requirements.txt như sau : 

annotated-types==0.7.0
anyio==4.9.0
click==8.1.8
colorama==0.4.6
dnspython==2.7.0
dotenv==0.9.9
exceptiongroup==1.2.2
fastapi==0.115.12
h11==0.16.0
idna==3.10
motor==3.7.0
pydantic==2.11.4
pydantic_core==2.33.2
pymongo==4.12.1
python-dotenv==1.1.0
python-slugify==8.0.4
sniffio==1.3.1
starlette==0.46.2
text-unidecode==1.3
typing-inspection==0.4.0
typing_extensions==4.13.2
uvicorn==0.34.2

Trong file main.py như sau :

from sys import prefix

from fastapi import FastAPI
from app.route import post as post_route
from app.route import video as video_route
import uvicorn

app = FastAPI()
app.include_router( post_route.router , prefix="/post",tags=['post'])
app.include_router( video_route.router , prefix="/video" ,tags=['video'])

@app.get("/")
def root_app():
    return { "status" : "active"}

#if __name__ == "__main__" :
 #   uvicorn.run("main:app" ,port=8888 )

mình cần lưu ý ở đây gọi đến route :

app.include_router( post_route.router , prefix="/post",tags=['post'])
app.include_router( video_route.router , prefix="/video" ,tags=['video'])

File postservice : 

from fastapi import HTTPException, Query
from app.helps.response import post_helper
from app.models.post import collection_name, post_collection
from app.schemas.post import PostCreate, PostUpdate
from bson import ObjectId
import motor.motor_asyncio
from pymongo import ReturnDocument

# Tạo bài viết mới
async def create_post(post: PostCreate):
    # Kiểm tra xem slug có tồn tại trong cơ sở dữ liệu không
    exit_post = await post_collection.find_one({"slug": post.slug})
    if exit_post:
        # Nếu tồn tại, trả về thông báo lỗi
        raise HTTPException(status_code=400, detail="Slug already exists")

    # Chèn bài viết vào cơ sở dữ liệu
    result = await post_collection.insert_one(post.dict())
    new_post = await post_collection.find_one({"_id": result.inserted_id})
    return post_helper(new_post)


# Cập nhật bài viết
async def update_post(post_id: str, post_update: PostUpdate):
    # Tìm bài viết theo id
    existing_post = await post_collection.find_one({"_id": ObjectId(post_id)})
    if not existing_post:
        # Nếu bài viết không tồn tại, trả về lỗi
        raise HTTPException(status_code=404, detail="Post not found")

    # Cập nhật bài viết với dữ liệu mới
    updated_post = await post_collection.find_one_and_update(
        {"_id": ObjectId(post_id)},
        {"$set": post_update.dict(exclude_unset=True)},  # Chỉ cập nhật các trường không rỗng
        return_document=ReturnDocument.AFTER # Trả về bài viết đã được cập nhật
    )
    return post_helper(updated_post)


# Xóa bài viết
async def delete_post(post_id: str):
    # Tìm bài viết theo id
    existing_post = await post_collection.find_one({"_id": ObjectId(post_id)})
    if not existing_post:
        # Nếu bài viết không tồn tại, trả về lỗi
        raise HTTPException(status_code=404, detail="Post not found")

    # Xóa bài viết
    await post_collection.delete_one({"_id": ObjectId(post_id)})

    return {"message": "Post deleted successfully"}


# Danh sách bài viết
async def list_posts(skip: int = 0, limit: int = 10):
    # Lấy danh sách bài viết, phân trang theo skip và limit
    posts_cursor = post_collection.find().skip(skip).limit(limit)
    posts = await posts_cursor.to_list(length=limit)
    return [post_helper(post) for post in posts]


# Tìm kiếm bài viết
async def search_posts(query: str, skip: int = 0, limit: int = 10):
    # Tìm bài viết theo tên hoặc slug
    search_query = {"$or": [
        {"name": {"$regex": query, "$options": "i"}},
        {"slug": {"$regex": query, "$options": "i"}}
    ]}

    posts_cursor = post_collection.find(search_query).skip(skip).limit(limit)
    posts = await posts_cursor.to_list(length=limit)
    return [post_helper(post) for post in posts]

File model : 

from app.models.database import db

collection_name = "posts"
post_collection = db[collection_name]

File database.py :

import os

from motor.motor_asyncio import AsyncIOMotorClient
from dotenv import load_dotenv
import os

load_dotenv()

MONGO_URL = os.getenv("MG_URL")
client = AsyncIOMotorClient(MONGO_URL)
db = client["fastapi"]

File .env :

MG_URL=mongodb+srv://xxx:xxxx@server1.xxxx.mongodb.net/?retryWrites=true&w=majority&appName=Server1

Bạn có thể download file để lấy demo